package com.wormpex.data.root.service.impl;

import com.wormpex.data.root.common.KafkaProducerUtil;
import com.wormpex.data.root.service.KafkaProducerService;
import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.errors.InterruptException;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;


/**
 * Created by ouryousa on 2017/8/16.
 */

@Service
public class KafkaProducerServiceImpl implements KafkaProducerService{

    private static final Logger logger = LoggerFactory.getLogger(KafkaProducerServiceImpl.class);

    private KafkaProducer<String,String> kafkaProducer;

    @PostConstruct
    public void init(){
        try {
            kafkaProducer = KafkaProducerUtil.getKafkaProducer();
            logger.info("Successfully init KafkaProducer");
        }catch(RuntimeException e){
            logger.error("Failed to init KafkaProducer");
        }
    }

    /**
     * 将value通过Kafka发送到topic
     * @param topic
     * @param value
     * @return
     */
    @Override
    public boolean sendMessage(String topic,String value) {
        ProducerRecord producerRecord = new ProducerRecord(topic,value);
        try {
            kafkaProducer.send(producerRecord);
        }catch (InterruptException e) {
            logger.error("Got InterruptException", e);
            return false;
        } catch (BufferExhaustedException e) {
            logger.error("Got BufferExhaustedException", e);
            return false;
        } catch (KafkaException e) {
            logger.error("Got KafkaException", e);
            return false;
        }catch(Throwable e){
            logger.error("Got Throwable", e);
            return false;
        }
        return true;
    }

}
